package com.erainm;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;

/**
 * Created by erainm on 2021/9/4 2:37 下午.
 *
 * @Description TODO: HBase 增删查数据，创建表，扫描等的基本API操作
 **/
public class HbaseClientDemo {
    Configuration conf = null;
    Connection conn = null;
    HBaseAdmin admin = null;

    @Before
    public void init() throws IOException {
        // 获取一个配置文件对象
        conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum","lg01,lg02");
        conf.set("hbase.zookeeper.property","2181");

        //通过conf获取到hbase集群的连接
        conn = ConnectionFactory.createConnection(conf);
    }

    /**
     * Created by erainm on 2021/9/4 2:54 下午.
     * @Description TODO: 创建一张hbase表
     **/
    @Test
    public void createTable() throws IOException {
        // 获取HBaseAdmin对象，用来创建表
        HBaseAdmin admin = (HBaseAdmin) conn.getAdmin();
        // 创建Htabledesc描述器，表描述器
        HTableDescriptor worker = new HTableDescriptor(TableName.valueOf("worker"));
        // 制定列族
        worker.addFamily(new HColumnDescriptor("info"));
        admin.createTable(worker);
        System.out.println("worker 表创建成功");
    }

    /**
     * Created by erainm on 2021/9/4 2:54 下午.
     * @Description TODO: 插入一条数据
     **/
    @Test
    public void putData() throws IOException {
        // 获取一个表对象
        Table worker = conn.getTable(TableName.valueOf("worker"));
        // 准备put对象，并指定rowkey
        Put put = new Put(Bytes.toBytes("110"));
        put.addColumn(Bytes.toBytes("info"),Bytes.toBytes("addr"),Bytes.toBytes("beijing"));
        // 插入数据，参数类型是put
        worker.put(put);
        // 准备list<puts> 可以执行批量插入数据
        // 关闭table对象
        worker.close();
        System.out.println("插入数据到worker表成功");
    }

    /**
     * Created by erainm on 2021/9/4 3:06 下午.
     * @Description TODO: 删除数据
     **/
    @Test
    public void deleteData() throws IOException {
        // 获取一个表对象
        Table worker = conn.getTable(TableName.valueOf("worker"));
        // 准备delete对象，并指定rowkey
        Delete delete = new Delete(Bytes.toBytes("110"));
        // 执行删除
        worker.delete(delete);
        // 关闭table对象
        worker.close();
        System.out.println("删除worker表中'110'数据成功");
    }

    /**
     * Created by erainm on 2021/9/4 3:14 下午.
     * @Description TODO: 查询数据
     **/
    @Test
    public void getData() throws IOException {
        // 获取一个表对象
        Table worker = conn.getTable(TableName.valueOf("worker"));
        // 获取Get对象
        Get get = new Get(Bytes.toBytes("110"));
        // 执行查询某个列族、活着某个列
        get.addFamily(Bytes.toBytes("info"));
        Result result = worker.get(get);
        // 获取到result中所有cell对象
        Cell[] cells = result.rawCells();
        // 遍历打印
        for (Cell cell : cells) {
            String rowkey = Bytes.toString(CellUtil.cloneRow(cell));
            String family = Bytes.toString(CellUtil.cloneFamily(cell));
            String qualifier = Bytes.toString(CellUtil.cloneQualifier(cell));
            String value = Bytes.toString(CellUtil.cloneValue(cell));

            System.out.println("rowkey: " + rowkey + " family: " + family + " qualifier: " + qualifier + " value: " + value);
        }
        // 关闭table对象
        worker.close();
        System.out.println("查询worker表中'110'数据成功");
    }

    /**
     * Created by erainm on 2021/9/4 3:20 下午.
     * @Description TODO: scan全表扫描
     **/
    @Test
    public void scanData() throws IOException {
        // 获取一个表对象
        Table worker = conn.getTable(TableName.valueOf("worker"));
        // 获取scan对象
        Scan scan = new Scan();
        // 执行扫描
        ResultScanner resultScanner = worker.getScanner(scan);

        for (Result result : resultScanner) {
            // 获取到result中所有cell对象
            Cell[] cells = result.rawCells();
            // 遍历打印
            for (Cell cell : cells) {
                String rowkey = Bytes.toString(CellUtil.cloneRow(cell));
                String family = Bytes.toString(CellUtil.cloneFamily(cell));
                String qualifier = Bytes.toString(CellUtil.cloneQualifier(cell));
                String value = Bytes.toString(CellUtil.cloneValue(cell));

                System.out.println("rowkey: " + rowkey + "; family: " + family + "; qualifier: " + qualifier + "; value: " + value);
            }
        }
        // 关闭table对象
        worker.close();
        System.out.println("scan worker表成功");
    }

    /**
     * Created by erainm on 2021/9/4 3:24 下午.
     * @Description TODO: 指定rowkey区间，范围扫描
     **/
    @Test
    public void scanStartEndData() throws IOException {
        // 获取一个表对象
        Table worker = conn.getTable(TableName.valueOf("worker"));
        // 获取scan对象
        Scan scan = new Scan();
        // 指定查询的rowkey区间,rowkey在hbase中是按字典序排序
        scan.setStartRow(Bytes.toBytes("001"));
        scan.setStopRow(Bytes.toBytes("2"));//110在这个区间，可以查到
        //scan.setStopRow(Bytes.toBytes("005"));//110不在这个区间，所以查不到
        // 执行扫描
        ResultScanner resultScanner = worker.getScanner(scan);

        for (Result result : resultScanner) {
            // 获取到result中所有cell对象
            Cell[] cells = result.rawCells();
            // 遍历打印
            for (Cell cell : cells) {
                String rowkey = Bytes.toString(CellUtil.cloneRow(cell));
                String family = Bytes.toString(CellUtil.cloneFamily(cell));
                String qualifier = Bytes.toString(CellUtil.cloneQualifier(cell));
                String value = Bytes.toString(CellUtil.cloneValue(cell));

                System.out.println("rowkey: " + rowkey + "; family: " + family + "; qualifier: " + qualifier + "; value: " + value);
            }
        }

        // 关闭table对象
        worker.close();
        System.out.println("scan worker表成功");
    }


    // 释放连接
    @After
    public void realse(){
        if (conn != null) {
            try {
                conn.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}
